#include "taskqueue.h"

taskqueue_t * tq_init(){
    debug_log("tq_init");
    taskqueue_t * tq = (taskqueue_t *)calloc(1,sizeof(taskqueue_t));
    tq->head = NULL;
    tq->rear = NULL;
    tq->size = 0;
    tq->exitflag = 0;
    int ret = pthread_mutex_init(&(tq->mutex),NULL);
    if(0 != ret)
        error_log("tq_init::pthread_mutex_init");
    ret = pthread_cond_init(&(tq->cond),NULL);
    if(0 != ret)
        error_log("tq_init::pthread_cond_init");
    return tq;
}
void tq_destory(taskqueue_t * queue){
    debug_log("tq_destory");
    int ret = pthread_mutex_lock(&queue->mutex);
    if(0 != ret)
        error_log("tq_destory::pthread_mutex_lock");
    if(0 == queue->exitflag){
        ret = pthread_mutex_unlock(&queue->mutex);
        if(0 != ret)
            error_log("tq_destory::pthread_mutex_unlock");
        return ;
    }
    while (0 != queue->size){
        ret = pthread_mutex_unlock(&queue->mutex);
        if(0 != ret)
            error_log("tq_destory::pthread_mutex_unlock");
        task_t * task;
        tq_dequeue(queue,task);
        free(task);
        ret = pthread_mutex_lock(&queue->mutex);
        if(0 != ret)
            error_log("tq_destory::pthread_mutex_lock");
    }
    ret = pthread_mutex_unlock(&queue->mutex);
    if(0 != ret)
        error_log("tq_destory::pthread_mutex_unlock");
    ret =  pthread_mutex_destroy(&queue->mutex);
    if(0 != ret)
        error_log("tq_destory::pthread_mutex_destory");
    ret = pthread_cond_destroy(&queue->cond);
    if(0 != ret)
        error_log("tq_destory::pthread_cond_destory");
    free(queue);
}
int tq_getsize(taskqueue_t * queue){
    return queue->size;
}
void tq_enqueue(taskqueue_t * queue,task_t * task){
    debug_log("tq_enqueue");
    int ret = pthread_mutex_lock(&queue->mutex);
    if(0 != ret)
        error_log("tq_enqueue::pthread_mutex_lock");
    if(0 != queue->exitflag){
        ret = pthread_mutex_unlock(&queue->mutex);
        if(0 != ret)
            error_log("tq_enqueue::pthread_mutex_unlock");
        return ;
    }
    
    //添加结点
    //空链表情况
    if(queue->size == 0){
        queue->head = task;
        queue->rear = task;
    }else//非空情况
    {
        queue->rear->next = task;
        queue->rear = NULL;
    }
    queue->size++;
    ret = pthread_mutex_unlock(&queue->mutex);
    if(0 != ret)
        error_log("tq_enqueue::pthread_mutex_unlock");
    pthread_cond_signal(&queue->cond);
}
void tq_dequeue(taskqueue_t * queue,task_t * task){
    debug_log("tq_dequeue");
    int ret = pthread_mutex_lock(&queue->mutex);
    if(0 != ret)
        error_log("tq_dequeue::pthread_mutex_lock");
    if(0 == queue->exitflag){
        debug_log("tq_dequeue::wait for new connect.");
        while(0 == queue->size){
            ret = pthread_cond_wait(&queue->cond,&queue->mutex);
        }
    }
    debug_log("tq_dequeue::new fd enqueue, dequue... ");
    task = queue->head;
    queue->head = task->next;
    task->next = NULL;
    if(queue->size ==1){
        queue->rear = NULL;   
    }
    queue->size--;
    ret = pthread_mutex_unlock(&queue->mutex);
    if(0 != ret)
        error_log("tq_dequeue::pthread_mutex_unlock");
}
void tq_setexit(taskqueue_t * queue){
    debug_log("tq_setexit");
    int ret = pthread_mutex_lock(&queue->mutex);
    if(0 != ret)
        error_log("tq_setexit::pthread_mutex_lock");
    queue->exitflag = 1;
    ret = pthread_mutex_unlock(&queue->mutex);
    if(0 != ret)
        error_log("tq_setexit::pthread_mutex_unlock");
}